package com._51doit.sparkwork

import java.sql.{Connection, DriverManager, PreparedStatement}

import com._51doit.spark05.IpUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object OrderMR2 {

  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName(this.getClass.getSimpleName)

    val sc: SparkContext = new SparkContext(conf)

    val ipdata: RDD[String] = sc.textFile("D:\\07spark\\spark-day04\\作业题\\ip.txt")
    val orderstr: RDD[String] = sc.textFile("D:\\07spark\\spark-day05\\作业题\\orders.log")

    val prodata: RDD[(Long, Long, String, String)] = ipdata.map(ida => {
      val datas: Array[String] = ida.split("\\|")
      val start = datas(2).toLong
      val end = datas(3).toLong
      val province = datas(6)
      val city: String = datas(7)
      // 起始ip地址 终止ip地址  省份
      (start, end, province, city)


    })
    
    /*
    规则： 稳定，需要长期维护； 频繁使用
    中间库 知识库
    collect收集
     */

    val ipRules: Array[(Long, Long, String, String)] = prodata.collect()


    val res1: RDD[((String, String), Int)] = orderstr.map(str => {
      val ordersdata: Array[String] = str.split(" ")
      val oid: String = ordersdata(0)
      val ip: String = ordersdata(1)
      val ptype: String = ordersdata(2)
      val product: String = ordersdata(3)
      val money: Int = ordersdata(4).toInt

      //将二进制的城市Ip转换为十进制
      val longip: Long = IpUtils.ip2Long(ip)

      //二分查找搜索省份
      val province: String = IpUtils.binarySearch(longip, ipRules.map(im => (im._1, im._2, im._3)))
      val city: String = IpUtils.binarySearch(longip, ipRules.map(imp => (imp._1, imp._2, imp._4)))

      ((province, city), money)

    })



    val res2: RDD[((String, String), Int)] = res1.reduceByKey(_+_)
    val res3: RDD[(String, Iterable[((String, String), Int)])] = res2.groupBy(_._1._1)

    val res4: RDD[(String, String)] = res3.mapValues(mt => {
      val list: List[(String, Int)] = mt.map({ case ((_, cty), mon) => (cty, mon) }).toList

      list.sortBy(-_._2).take(3).mkString("")


    })


    res4.foreach(println)





    /*res1.foreachPartition(rf=>{

      //建立与数据库的链接 传统的加载
      val url = "jdbc:mysql://127.0.0.1:3306/guanxi?characterEncoding=utf-8"

      var conn: Connection = null
      var pst1: PreparedStatement = null
      var pst2: PreparedStatement = null

      try {
        conn = DriverManager.getConnection(url, "root", "123456")
        //创建表
        pst1 = conn.prepareStatement("create table if not exists orders0 (province varchar(30),cnts int)")
        pst1.execute()
        //插入数据语句
        pst2 = conn.prepareStatement("insert into orders0 values(?,?)")

        rf.foreach({
          case (pro, mon) => {
            pst2.setString(1, pro)
            pst2.setInt(2, mon)
            pst2.execute()
          }

        })
      } catch {
        //打印错误
        case e:Exception => e.printStackTrace()
      } finally {
        IpUtils.closeconn(conn,pst1,pst2)
      }



    })
*/
    sc.stop()








  }







}
